package com.weishi.pay.serviceImpl;



import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.weishi.pay.api.exception.PayErrors;
import com.weishi.pay.api.service.RocketMqService;
import com.weishi.pay.common.IdWorker;
import com.weishiinfo.commons.exception.RawErrorCodeException;

/**
 * Class_Name: RocketMqServiceImpl
 * Package: com.weishinfo.center.user.rpc.service.impl
 * Describer: RocketMq消息服务实现类
 *
 * @author chixiahua
 * @date 2018/7/4 15:16
 **/
@Service
public class RocketMqServiceImpl implements RocketMqService {

    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMqServiceImpl.class);

    @Autowired
    private DefaultMQProducer userMQProducer;

    @Override
    public void sendMsg(String topic, String tag, String body) throws RawErrorCodeException {
        // 生成消息唯一标示
        IdWorker idWorker = new IdWorker(1, 1);
        String key = idWorker.nextIdStr();
        Message message = buildMessage(topic, tag, key, body);
        SendResult sendResult;
        // 发送消息
        try {
            //sendResult = userMQProducer.send(message);
            LOGGER.info("########send begin");
            userMQProducer.send(message, new SendCallback() {

                @Override
                public void onSuccess(SendResult sendResult) {
                    LOGGER.info("###############sendMQ ok:"+sendResult);
                    // 发送成功处理
                    //按照key查找错误表是否已经存在
                    //VoMqSendError voMqSendError = mqSendErrorService.selectByTopicAndAndTagsAndKeys(topic, tags, keys);
                    //if(null != voMqSendError){
                        //如果存在错误表中，则移历史
                    //    mqSendErrorService.moveToHis(voMqSendError);
                   // }
                }

                @Override
                public void onException(Throwable e) {
                    //System.out.println(e);
                    LOGGER.error("###########sendMQ error:"+e.getStackTrace());
                    // 发送失败处理
                    //processError(topic, tags, keys, msgBody, e.getMessage());
                    throw PayErrors.MQ_SENDERROR.bornException();
                }
            });

        } catch (Exception e){//(MQClientException | RemotingException | InterruptedException | MQBrokerException e) {
            LOGGER.info("########send error");
            e.printStackTrace();
            throw PayErrors.MQ_SENDERROR.bornException();

        }
        LOGGER.info("########send end");
        // 发送失败
        //if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
        //    throw new PayCenterException(ErrorMessageEnums.MQ_SENDERROR);
        //}

    }

    /**
     * 构建消息
     *
     * @param topic topic
     * @param tag   tag
     * @param key   key
     * @param body  body
     * @return Message 消息对象
     * @throws PayCenterException PayCenterException
     */
    private Message buildMessage(String topic, String tag, String key, String body) throws RawErrorCodeException {
        return new Message(topic, tag, key, body.getBytes());
    }
}
